package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author WEIYUNHUI
 * @date 2023/6/10 11:15
 *
 * 并行度: 并行执行的程度。
 *   算子的并行度:   某个算子(map、 flatMap 、 filter)的并行程度
 *   作业的并行度:   作业的并行度由算子的并行度来决定。 作业的并行度取决于当前作业中所有算子中并行度最高的算子。
 *
 * 并行度的设置:
 *    1. 在IDEA中运行，如果不指定并行度， 默认取CPU支持的最大的线程数.
 *    2. 给每个算子单独设置并行度 , 非并行算子的并行度只能是1 。  xx.map().setParallelism(1);
 *    3. 设置全局并行度 env.setParallelism(1) ;
 *    4. 提交作业设置并行度
 *         webUI中指定
 *         命令行通过-p参数来指定。
 *    5. flink-conf.yaml配置文件中设置并行度。 默认是1 。
 *
 *
 * 并行度设置优先级:
 *    代码中给算子单独设置  >  代码中全局设置  > 提交作业设置并行度  >  配置文件中设置并行度
 *
 * 建议:
 *   提交作业的时候动态指定并行度。
 *
 *
 */
public class Flink08_Parallelism {
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port" , 5678);

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        //设置并行度
        env.setParallelism(2) ;

        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888) ;
                //.setParallelism(1);

        // 3.转换处理
        // 3.1 切分数据， 处理成(word, 1)
        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMapDs = ds.flatMap(
                new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                        String[] words = line.split(" ");
                        for (String word : words) {
                            out.collect(Tuple2.of(word, 1));
                        }

                    }
                }
        );//.setParallelism(3);
        // 3.2 按照单词分组
        KeyedStream<Tuple2<String, Integer>, String> keyByDs = flatMapDs.keyBy(
                new KeySelector<Tuple2<String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Integer> value) throws Exception {
                        return value.f0; // 使用tuple中的第一个元素作为key
                    }
                }
        );
        // 3.3 汇总
        // 使用Tuple中的第二个元素进行汇总
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumDs = keyByDs.sum(1);//.setParallelism(4);

        // 4. 输出结果
        sumDs.print();//.setParallelism(5);

        // 5. 启动执行
        env.execute();
    }

}


















